"""
Copyright (c) 2009 James Lindstorff
All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:
1. Redistributions of source code must retain the above copyright
   notice, this list of conditions and the following disclaimer.
2. Redistributions in binary form must reproduce the above copyright
   notice, this list of conditions and the following disclaimer in the
   documentation and/or other materials provided with the distribution.

THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED.  IN NO EVENT SHALL AUTHOR OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
SUCH DAMAGE.
"""

import re
import time
import unittest

from varnishmport import CLIS_UNKNOWN
from varnishmport import CLIS_OK
from varnishmport import CLIS_CANT
from varnishmport import CLIS_CLOSE
from varnishmport import CLIS_PARAM
from varnishmport import CLIS_TOOFEW
from varnishCLI import VarnishCLI


SLEEP_TIME = 0.5 # Time in seconds to wait on state changes (start and stop)

REMOTE_VCL = '/usr/local/etc/varnish/default.vcl' # Need to correct this to actual system.

class TestVarnishCLI(unittest.TestCase):
    """Tests for VarnishCLI.
    The prerequisits to running these tests are a running version of varnishd
    and the constructor in setUp corrected to the installation to run against.
    It's easist to run agaist a default installation because some of the
    tests do tests against default values for instance default_ttl. If running
    against a default.vcl make sure to enable the backend definition. Also
    you should adjust REMOTE_VCL, SLEEP_TIME to match actual conditions. """

    def setUp(self):
        self.cli = VarnishCLI("localhost", 6082, "/usr/local/etc/varnish/secret")
        # let things settle before proceding
        time.sleep(0.5)

    def tearDown(self):
        if not self.cli is None:
            self.cli.quit()
            self.cli = None


    def __ensureCacheProcessStarted(self):
        if not self.cli.cacheProcessRunning():
            (sc, txt) = self.cli.start();
            self.assertEquals(CLIS_OK, sc)
            time.sleep(SLEEP_TIME)

    def __ensureCacheProcessStopped(self):
        if self.cli.cacheProcessRunning():
            (sc, txt) = self.cli.stop();
            self.assertEquals(CLIS_OK, sc)
            time.sleep(SLEEP_TIME)


    def testQuit(self):
        (sc, txt) = self.cli.quit()
        self.assertEquals(CLIS_CLOSE, sc, "Didn't get a close status code")
        self.cli = None # to avoid tearDown opening and closing just to send the quit


    def testStatus(self):
      (sc, txt) = self.cli.status()
      self.assertEquals(CLIS_OK, sc)


    def testBanner(self):
        expected = "Varnish HTTP accelerator CLI\."
        (sc, txt) = self.cli.banner()
        match = re.search(expected, txt)
        self.assertFalse(match is None, "Didn't match banner")


    def testStartAlreadyStarted(self):
        self.__ensureCacheProcessStarted()
        # Try send start when is started state
        (sc, txt) = self.cli.start()
        self.assertEquals(CLIS_CANT, sc)
        self.assertEquals("Child in state running", txt)
        time.sleep(SLEEP_TIME)


    def testStartStop(self):
        self.__ensureCacheProcessStopped()
        # State at this point is stopped, so we try to start
        (sc, txt) = self.cli.start();
        self.assertEquals(CLIS_OK, sc)
        time.sleep(SLEEP_TIME)
        # Check we are running
        self.assertTrue(self.cli.cacheProcessRunning(), "Process not started")
        # State is running so we stop
        (sc, txt) = self.cli.stop();
        self.assertEquals(CLIS_OK, sc)
        time.sleep(SLEEP_TIME)
        # Check we are actually stopped
        self.assertFalse(self.cli.cacheProcessRunning(), "Process not stopped")
        # Start up the child process again for good measure
        (sc, txt) = self.cli.start();
        self.assertEquals(CLIS_OK, sc)
        time.sleep(SLEEP_TIME)


    def testPing(self):
        (sc, txt, timestamp, pongtime, duration) = self.cli.ping()
        self.assertEquals(CLIS_OK, sc)
        self.assertTrue(txt.startswith("PONG"))
        self.assertTrue(0 < timestamp)
        self.assertFalse(pongtime is None)
        self.assertFalse(duration is None)


    def testPingWithTimestamp(self):
        timestampsend = int(time.time())
        (sc, txt, timestamp, pongtime, duration) = self.cli.ping(timestampsend)
        self.assertEquals(CLIS_OK, sc)
        self.assertTrue(txt.startswith("PONG"))
        self.assertEquals(timestampsend, timestamp)
        self.assertFalse(pongtime is None)
        self.assertFalse(duration is None)


    def testHelpList(self):
        (sc, commandlist) = self.cli.help()
        self.assertEquals(CLIS_OK, sc)
        self.assertTrue(0 < len(commandlist))


    def testHelpUnknownCommand(self):
        (sc, text, description) = self.cli.help("fisk")
        self.assertEquals(CLIS_UNKNOWN, sc)
        self.assertEquals("Unknown command.", description)


    def testHelpSpecific(self):
        (sc, command, description) = self.cli.help("vcl.list")
        self.assertEquals(CLIS_OK, sc)
        self.assertEquals("vcl.list", command)
        self.assertEquals("List all loaded configuration.", description)


    def testHelpDetail(self):
        detail = self.cli.helpDetail()
        self.assertTrue(0 < len(detail))
        (commandtext, description) = detail["vcl.list"]
        self.assertEqual("vcl.list", commandtext)
        self.assertEquals("List all loaded configuration.", description)


    def testStats(self):
        (sc, stats) = self.cli.stats()
        self.assertEquals(CLIS_OK, sc)
        self.assertTrue(0 < len(stats))


    def testParamShowAll(self):
        (sc, params) = self.cli.paramShow()
        self.assertEquals(CLIS_OK, sc)


    def testParamShowSpecific(self):
        (sc, params) = self.cli.paramShow('default_ttl')
        self.assertEquals(CLIS_OK, sc)
        self.assertEquals('default_ttl', params[0][0])
        self.assertTrue(params[0][1].startswith('Default is 120'))
        self.assertEquals('120 [seconds]', params[0][2])


    def testParamShowSpecificUnkonwn(self):
        (sc, params) = self.cli.paramShow('unknownparameter')
        self.assertEquals(CLIS_PARAM, sc)
        self.assertEquals(0, len(params))


    def testParamSet(self):
        (sc, txt) = self.cli.paramSet('default_ttl', 3600)
        self.assertEquals(CLIS_OK, sc)
        (sc, params) = self.cli.paramShow('default_ttl')
        self.assertEquals(CLIS_OK, sc)
        self.assertEquals('3600 [seconds]', params[0][2])
        (sc, txt) = self.cli.paramSet('default_ttl', 120)
        self.assertEquals(CLIS_OK, sc)


    def testParamSetUnknownParameter(self):
        (sc, txt) = self.cli.paramSet('fish', 42)
        self.assertEquals(CLIS_PARAM, sc)


    def testParamSetMissingValue(self):
        (sc, txt) = self.cli.paramSet('default', '')
        self.assertEquals(CLIS_TOOFEW, sc)


    def __purgeListHasTest(self, test):
        (sc, purges) = self.cli.purgeList()
        self.assertEquals(CLIS_OK, sc)
        self.assertTrue(0 < len(purges))
        for purgeitem in purges:
            if purgeitem[4] ==  test:
                return True
        return False


    def testPurge(self):
        condition = 'req.url ~ fisk && req.url ~ hest'
        (sc, txt) = self.cli.purge(condition)
        self.assertEquals(CLIS_OK, sc)
        self.assertTrue(self.__purgeListHasTest(condition))


    def testPurgeHash(self):
        # make sure we can do a purge.hash
        (sc, params) = self.cli.paramShow('purge_hash')
        self.assertEquals(CLIS_OK, sc)
        self.assertEquals(1, len(params))
        purgehash = params[0][2]
        if purgehash.startswith('off'):
            (sc, txt) = self.cli.paramSet('purge_hash', 'on')
            self.assertEquals(CLIS_OK, sc)
            # requires restart to take effect to we restart
            self.cli.restart(SLEEP_TIME)
        (sc, txt) = self.cli.purgeHash('^9')
        self.assertEquals(CLIS_OK, sc)
        # Check it's in the active list.
        self.assertTrue(self.__purgeListHasTest('obj.hash ~ ^9'))
        # restore hash purge default
        if purgehash.startswith('off'):
            (sc, txt) = self.cli.paramSet('purge_hash', 'off')
            self.assertEquals(CLIS_OK, sc)
            self.cli.restart(SLEEP_TIME)


    def testPurgeUrl(self):
        (sc, txt) = self.cli.purgeUrl('^/fish/.+jsp$')
        self.assertEquals(CLIS_OK, sc)
        self.assertTrue(self.__purgeListHasTest('req.url ~ ^/fish/.+jsp$'))
        

    def testPurgeList(self):
        # Also tested by testPurgeHash and testPurgeUrl that's why this is thin
        (sc, purges) = self.cli.purgeList()
        self.assertEquals(CLIS_OK, sc)


    def testRestart(self):
        (sc, txt) = self.cli.restart()
        self.assertEquals(CLIS_OK, sc)


    def __activeVcl(self):
        (sc, vcls) = self.cli.vclList()
        for vcl in vcls:
            if vcl[0] == 'active':
                return vcl[2]
        return None



    def testVclList(self):
        states = ('discarded', 'available', 'active')
        (sc, vcls) = self.cli.vclList()
        self.assertEquals(CLIS_OK, sc)
        self.assertTrue(0 < len(vcls))
        for vcl in vcls:
            self.assertTrue(vcl[0] in states, vcl[0])


    def testVclLoad(self):
        configname = 'testVclLoad'
        # Try to discard testVclLoad is discarded
        (sc, txt) = self.cli.vclDiscard(configname)
        self.assertTrue(CLIS_OK == sc or CLIS_PARAM == sc)
        (sc, txt) = self.cli.vclLoad(configname, REMOTE_VCL)
        self.assertEquals(CLIS_OK, sc)
        # load again for used name test
        (sc, txt) = self.cli.vclLoad(configname, REMOTE_VCL)
        self.assertEquals(CLIS_PARAM, sc)
        (sc, txt) = self.cli.vclDiscard(configname)
        self.assertEquals(CLIS_OK, sc)
        self.cli.restart()
        
        
    def testVclDiscardActive(self):
        active = self.__activeVcl()
        (sc, txt) = self.cli.vclDiscard(active)
        self.assertEquals(CLIS_PARAM, sc)


    def testVclInline(self):
        configname = 'inlined'
        # make sure configuration isn't loaded already
        active = self.__activeVcl()
        if active == configname:
            self.fail("Can't run test because the active configuration is our test name")
        (sc, txt) = self.cli.vclDiscard(configname)
        # load the test vcl and inline it.
        vclFile = open('py.vcl', 'r')
        vcl = vclFile.read()
        vclFile.close()
        (sc, txt) = self.cli.vclInline(configname, vcl)
        self.assertEquals(CLIS_OK, sc, txt)
        # avtivate the inlined vcl.
        (sc, txt) = self.cli.vclUse(configname)
        self.assertEquals(CLIS_OK, sc)
        # is it really active
        newactive = self.__activeVcl()
        self.assertEquals(newactive, configname)
        # Back to original
        (sc, txt) = self.cli.vclUse(active)
        self.assertEquals(CLIS_OK, sc)
        # discard the inline version
        (sc, txt) = self.cli.vclDiscard(configname)
        self.assertEquals(CLIS_OK, sc)
        # restart to clear
        self.cli.restart()


    def testVclShowActive(self):
        active = self.__activeVcl()
        (sc, vcl) = self.cli.vclShow(active)
        self.assertEquals(CLIS_OK, sc)
        self.assertFalse(-1 == vcl.find("backend"))


    def testDebugHealth(self):
        configname = 'healthtesting'
        # make sure configuration isn't loaded already
        (sc, txt) = self.cli.vclDiscard(configname)
        # load the test vcl and inline it.
        vclFile = open('py.vcl', 'r')
        vcl = vclFile.read()
        vclFile.close()
        (sc, txt) = self.cli.vclInline(configname, vcl)
        self.assertEquals(CLIS_OK, sc, txt)
        # at this point we have the backends defined with probing
        (sc, health) = self.cli.health()
        self.assertEquals(CLIS_OK, sc)
        self.assertTrue(0 < len(health))
        possibleStates = ['Healthy', 'Sick']
        for healthitem in health:
            self.assertFalse(healthitem[0] is None)
            self.assertTrue(healthitem[1] in possibleStates)
            self.assertTrue(-1.0 < healthitem[2])


    def testLastProperties(self):
        (sc, txt) = self.cli.status()
        self.assertEquals(CLIS_OK, sc)
        self.assertEquals(sc, self.cli.statuscode)
        self.assertEquals(txt, self.cli.response)
        self.assertEquals('status', self.cli.command)


if __name__ == "__main__":
    suite = unittest.TestLoader().loadTestsFromTestCase(TestVarnishCLI)
    unittest.TextTestRunner(verbosity=2).run(suite)
    